package com.atguigu.edu.realtime.app.dwd;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeOrderCancelDetail {//取消订单表   dwd_trade_order_cancel_detail  **************

        public static void main(String[] args) {
            // 0 环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(4);
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
            tableEnv.getConfig().setIdleStateRetention(Duration.ofMillis(20*60000)); //生命周期

            //  1  获得topic_db
            String createTopicDbSQL="     CREATE TABLE topic_db (\n" +
                    "        `database` STRING,\n" +
                    "       `table` STRING,\n" +
                    "        `type` STRING,\n" +
                    "         `data` MAP<STRING,STRING>,\n" +
                    "         `old` MAP<STRING,STRING> ,\n" +
                    "          `proc_time`  as proctime(), \n"+
                    "         ts   STRING \n"+
                    "          ) WITH (\n" +
                    "          'connector' = 'kafka',\n" +
                    "          'topic' = 'topic_db',\n" +
                    "          'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                    "          'properties.group.id' = 'dwd_trade_order_detail',\n" +
                    "          'scan.startup.mode' = 'group-offsets',\n" +
                    "           'format' = 'json'\n" +
                    "         )";
            tableEnv.executeSql(createTopicDbSQL);

            //  2  分别创建四个动态表
            //2.1   订单主表      order_info
            Table orderInfo = tableEnv.sqlQuery("select \n" +
                    "data['id'] id,\n" +
                    "data['user_id'] user_id,\n" +
                    "data['province_id'] province_id\n" +
                    "from `topic_db`\n" +
                    "where `table` = 'order_info'\n" +
                    "and `type` = 'update' and data['order_status']='1003'");
            tableEnv.createTemporaryView("order_info", orderInfo);

            //2.2   订单明细表  order_detail
            Table orderDetail = tableEnv.sqlQuery("select \n" +
                    "data['id'] id,\n" +
                    "data['course_id'] course_id,\n" +
                    "data['course_name'] course_name,\n" +
                    "data['order_id'] order_id,\n" +
                    "data['user_id'] user_id,\n" +
                    "data['origin_amount'] origin_amount,\n" +
                    "data['coupon_reduce'] coupon_reduce,\n" +
                    "data['final_amount'] final_amount,\n" +
                    "data['create_time'] create_time,\n" +
                    "data['update_time'] update_time,\n" +
                    "ts,\n" +
                    "proc_time\n" +
                    "from `topic_db` where `table` = 'order_detail' " +
                    "and `type` = 'insert'\n");
            tableEnv.createTemporaryView("order_detail", orderDetail);

            // 3   字典维度表
            String baseDicDDL = MySqlUtil.getBaseDicDDL();
            tableEnv.executeSql(baseDicDDL);

            //4   进行join操作
            Table resultTable = tableEnv.sqlQuery("select \n" +
                    "od.id,\n" +
                    "od.course_id,\n" +
                    "oi.user_id,\n" +
                    "od.course_name,\n" +
                    "od.order_id,\n" +
                    "od.origin_amount,\n" +
                    "od.coupon_reduce,\n" +
                    "od.final_amount,\n" +
                    "od.create_time,\n" +
                    "od.update_time,\n" +
                    "od.ts\n" +
                    "from order_detail od \n" +
                    "join order_info oi\n" +
                    "on od.order_id = oi.id");
            tableEnv.createTemporaryView("result_table", resultTable);
            //   tableEnv.executeSql("select * from "+resultTable).print();

            // 5  目标表的定义
            tableEnv.executeSql("" +
                    "create table dwd_trade_order_cancel_detail(\n" +
                    "id string,\n" +
                    "course_id string,\n" +
                    "user_id string,\n" +
                    "course_name string,\n" +
                    "order_id string,\n" +
                    "origin_amount string,\n" +
                    "coupon_reduce string,\n" +
                    "final_amount string,\n" +
                    "create_time string,\n" +
                    "update_time string,\n" +
                    "ts string,\n" +
                    "primary key(id) not enforced\n" +
                    ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_cancel_detail"));

            // 6   写入目标表
            tableEnv.executeSql("" +
                    "insert into dwd_trade_order_cancel_detail \n" +
                    "select * from result_table");
    }
}
